package com.ai.nsa.actions;

import com.ai.nsa.NSACache;
import com.ai.nsa.utils.CalendarUtils;
import com.ai.nsa.utils.KeyUtils;
import com.ai.nsa.utils.LeftPad;
import com.ai.nsa.utils.hbase.HBaseOperations;
import com.ai.nsa.utils.hbase.model.HBaseRow;
import com.ai.nsa.utils.lock.LockFactory;
import com.ai.nsa.utils.page.model.HBasePage;
import com.ai.nsa.utils.page.model.Page;
import com.ai.nsa.utils.partition.HashPartitioner;
import com.ai.nsa.utils.partition.Partitioner;
import com.asiainfo.appserver.Action;
import com.asiainfo.appserver.Request;
import com.asiainfo.appserver.Response;
import com.asiainfo.appserver.annotations.Function;
import com.asiainfo.appserver.annotations.Parameter;
import com.asiainfo.appserver.annotations.Type;
import org.apache.commons.lang.StringUtils;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.filter.CompareFilter.CompareOp;
import org.apache.hadoop.hbase.filter.Filter;
import org.apache.hadoop.hbase.filter.FilterList;
import org.apache.hadoop.hbase.filter.SingleColumnValueFilter;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.Map;

import static com.ai.nsa.NSAConstants.*;

/**
 * @author Feng Chen 2017/6/8.
 */
@Function(description = "日志检索(列表)", parameters = {
        @Parameter(name = "SERIAL_NUMBER", type = Type.String, description = "手机号码", required = true),
        @Parameter(name = "START_DATE", type = Type.String, description = "开始时间", required = true),
        @Parameter(name = "END_DATE", type = Type.String, description = "结束时间", required = true),
        @Parameter(name = "IN_MODE_CODE", type = Type.String, description = "接入渠道", required = false),
        @Parameter(name = "INTERACT_TYPE", type = Type.String, description = "交互类型", required = false),
        @Parameter(name = "PAGE_NO", type = Type.String, defaultValue = "1", description = "当前页", required = false),
        @Parameter(name = "PAGE_SIZE", type = Type.String, defaultValue = "3", description = "页大小", required = false)
})
@Component
public class NSAQry implements Action {

    @Autowired
    private HBaseOperations hbase;

    @Autowired
    private NSACache cache;

    private final static Logger logger = LoggerFactory.getLogger(NSAQry.class);

    public Response execute(final Request request) throws Exception {

        String serialNumber = request.getString("SERIAL_NUMBER");
        String startDate = request.getString("START_DATE");
        String endDate = request.getString("END_DATE");
        String inModeCode = request.getString("IN_MODE_CODE");
        String interactType = request.getString("INTERACT_TYPE");
        String pageNo = request.getString("PAGE_NO");
        String pageSize = request.getString("PAGE_SIZE");

        try {

            Page<HBaseRow> page = fetch(serialNumber, startDate, endDate, inModeCode, interactType, pageNo, pageSize);

            return result(page, pageNo);
        } catch (Exception e) {
            logger.error("", e);
            return Response.fail(e);
        }
    }

    private Response result(Page<HBaseRow> page, String pageNo) {

        Response response = Response.success();

        int current = Integer.parseInt(pageNo);

        if (page == null) {
            response.put("totalPage", 0);
            response.put("totalCount", 0);
            response.put("current", current);
            response.put("rows", Collections.emptyList());
        } else {
            response.put("totalPage", page.getTotalPage());
            response.put("totalCount", page.getTotalCount());
            response.put("current", current);
            response.put("rows", extract(page.page(current)));
        }

        return response;
    }

    private List<Map<String, String>> extract(List<HBaseRow> rows) {
        return extract(rows, new Extracter<List<HBaseRow>, List<Map<String, String>>>() {

            @Override
            public List<Map<String, String>> extract(List<HBaseRow> rows) {
                List<Map<String, String>> result = new ArrayList<Map<String, String>>();
                if (rows != null && !rows.isEmpty()) {
                    for (HBaseRow row : rows) {
                        Map<String, String> res = row.getFamily(HBASE_FAMILY);
                        res.put("ROW_KEY", row.getRowKey());
                        result.add(res);
                    }
                }
                return result;
            }

        });
    }


    private List<Map<String, String>> extract(List<HBaseRow> rows, Extracter<List<HBaseRow>, List<Map<String, String>>> extracter) {
        return extracter.extract(rows);
    }

    private Page<HBaseRow> fetch(
            String serialNumber, String startDate, String endDate,
            String inModeCode, String interactType, String pageNo, String pageSize
    ) throws Exception {
        PageFetcher fetcher = new PageFetcher();
        return fetcher.fetch(serialNumber, startDate, endDate, inModeCode, interactType, pageNo, pageSize);
    }

    /////////////////////////////////////////////////


    private class PageFetcher implements Fetcher<HBasePage> {

        @Override
        public HBasePage fetch(
                String serialNumber, String startDate, String endDate,
                String inModeCode, String interactType, String pageNo, String pageSize
        ) throws Exception {
            synchronized (LockFactory.getLock(serialNumber)) {
                HBasePage page = get(serialNumber, startDate, endDate, inModeCode, interactType, pageSize);
                if (page == null) {

                    logger.info("NOT FOUND CACHE, KEY: {}", Key.get(serialNumber, startDate, endDate, inModeCode, interactType, pageSize));

                    // 生成缓存页
                    PageGenerater generater = new PageGenerater();
                    generater.generate(
                            serialNumber, startDate, endDate, inModeCode,
                            interactType, pageNo, pageSize
                    );
                }
                return get(serialNumber, startDate, endDate, inModeCode, interactType, pageSize);
            }
        }

        // SECTION: INNER HELPER

        private HBasePage get(
                String serialNumber, String startDate, String endDate,
                String inModeCode, String interactType, String pageSize
        ) throws Exception {
            return (HBasePage) cache.get(Key.get(serialNumber, startDate, endDate, inModeCode, interactType, pageSize));
        }

    }

    private class HbaseScanner implements Scanner<List<HBaseRow>> {

        private Partitioner partitioner = new HashPartitioner();

        @Override
        public List<HBaseRow> scan(
                String serialNumber, String startDate, String endDate,
                String inModeCode, String interactType, String pageNo, String pageSize
        ) throws Exception {
            Scan scan = getScan(serialNumber, startDate, endDate);
            scan.setFilter(packageFilters(inModeCode, interactType, startDate, endDate));
            int duration = CalendarUtils.getMonthDuration(startDate, endDate);

            List<HBaseRow> rows = new ArrayList<HBaseRow>();

            for (int i = 0; i <= duration; i++) {
                int month = CalendarUtils.getMonth(startDate, i);
                String table = HBASE_MASTER_TABLE + UNDERLINE + LeftPad.pad(month, 2, "0");
                List<HBaseRow> rs = hbase.scan(table, scan);
                if (rs != null && rs.size() > 0) {
                    rows.addAll(rs);
                }
            }

            return rows;
        }

        private Scan getScan(String serialNumber, String startDate, String endDate) {
            Scan scan = new Scan();
            int part = partitioner.getPartition(serialNumber);
            scan.setStartRow((LeftPad.pad(part, 2, "0") + serialNumber + startDate).getBytes());
            scan.setStopRow((LeftPad.pad(part, 2, "0") + serialNumber + endDate).getBytes());
            return scan;
        }

        private FilterList packageFilters(String inModeCode, String interactType, String startDate, String endDate) {

            // MUST_PASS_ALL(条件 AND) MUST_PASS_ONE（条件OR）
            FilterList filterList = new FilterList(FilterList.Operator.MUST_PASS_ALL);

            if (StringUtils.isNotEmpty(inModeCode)) {
                filterList.addFilter(newFilter(HBASE_FAMILY, COLUMN_IN_MODE_CODE, CompareOp.EQUAL, inModeCode));
            }

            if (StringUtils.isNotEmpty(interactType)) {
                filterList.addFilter(newFilter(HBASE_FAMILY, COLUMN_INTERACT_TYPE, CompareOp.EQUAL, interactType));
            }

            filterList.addFilter(newFilter(HBASE_FAMILY, COLUMN_TRADE_DATE, CompareOp.GREATER_OR_EQUAL, startDate));
            filterList.addFilter(newFilter(HBASE_FAMILY, COLUMN_TRADE_DATE, CompareOp.LESS, endDate));

            return filterList;
        }


        private Filter newFilter(String family, String column, CompareOp op, String value) {
            return new SingleColumnValueFilter(family.getBytes(), column.getBytes(), op, value.getBytes());
        }

    }

    private class PageGenerater {

        public synchronized void generate(
                String serialNumber, String startDate, String endDate, String inModeCode,
                String interactType, String pageNo, String pageSize
        ) throws Exception {

            HbaseScanner scanner = new HbaseScanner();
            List<HBaseRow> rows = scanner.scan(serialNumber, startDate, endDate, inModeCode, interactType, pageNo, pageSize);

            HBasePage page = new HBasePage(rows, Integer.parseInt(pageSize));
            String key = Key.get(serialNumber, startDate, endDate, inModeCode, interactType, pageSize);

            // 保存到缓存中
            cache(serialNumber, key, page);
        }


        private void cache(String serialNumber, String key, HBasePage page) throws Exception {

            // 保存索引
            logger.info("cache index : {}", serialNumber);
            cache.push(serialNumber, key);

            // 保存数据
            logger.info("cache data , key : {}, totalPage: {}", key, page.getTotalPage());
            cache.put(key, page);

        }

    }

    public static abstract class Key {

        private Key() {
            // prevent instantiation
        }

        public static String get(
                String serialNumber, String startDate, String endDate,
                String inModeCode, String interactType, String pageSize
        ) {
            return KeyUtils.get(
                    serialNumber, startDate, endDate,
                    LeftPad.pad(inModeCode, 2, "0"),
                    LeftPad.pad(interactType, 4, "0"),
                    LeftPad.pad(pageSize, 6, "0")
            );
        }

    }

    private interface Fetcher<T> {

        T fetch(
                String serialNumber, String startDate, String endDate, String inModeCode, String interactType,
                String pageNo, String pageSize
        ) throws Exception;

    }

    private interface Extracter<F, T> {

        T extract(F f);

    }

    private interface Scanner<T> {

        T scan(
                String serialNumber, String startDate, String endDate, String inModeCode, String interactType,
                String pageNo, String pageSize
        ) throws Exception;

    }


}